package com.shujia.core

import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala._


object Demo8Checkpoint {
  def main(args: Array[String]): Unit = {
    /**
      * 实时计算：7*24小时不断计算，如果任务失败了，重新启动，怎么才能保证之前的计算结果不丢失
      *
      * checkpoint: 定时将flink计算的状态持久化到hdfs中，
      * 如果任务执行失败可以恢复到checkpoint的位置重新计算，不会丢失之前的计算结果
      *
      * checkpoint在本地不能使用，需要在集群中使用
      *
      */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    // 每 1000ms 开始一次 checkpoint
    env.enableCheckpointing(5000)

    // 高级选项：

    // 设置模式为精确一次 (这是默认值)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    // 确认 checkpoints 之间的时间会进行 500 ms
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

    // Checkpoint 必须在一分钟内完成，否则就会被抛弃
    env.getCheckpointConfig.setCheckpointTimeout(60000)

    // 允许两个连续的 checkpoint 错误
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2)

    // 同一时间只允许一个 checkpoint 进行
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

    // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
    env.getCheckpointConfig.setExternalizedCheckpointCleanup(
      ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)


    /**
      * 设置checkpoint保存数据方式和位置
      *
      */
    //老版本
    //env.setStateBackend(new RocksDBStateBackend("hdfs://master:9000/flink/checkpoint/"))

    //新版本
    //rocksDB状态后端
    env.setStateBackend(new EmbeddedRocksDBStateBackend(true))
    //env.setStateBackend(new HashMapStateBackend())
    env.getCheckpointConfig.setCheckpointStorage("hdfs://master:9000/flink/checkpoint/")


    /**
      *
      * 任务失败之后可以从上一次成功的checkpiint的位置恢复任务
      * -s ： 指定从哪一个位置恢复任务
      *
      * flink run -t yarn-session -Dyarn.application.id=application_1654852007237_0001
      * -c com.shujia.core.Demo8Checkpoint
      * -s hdfs://master:9000/flink/checkpoint/dfb2be118dd1c44511793cada7c1fb2d/chk-64
      * flink-1.0.jar
      *
      */


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val wordsDS: DataStream[String] = linesDS.flatMap(_.split(","))

    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))

    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)

    countDS.print()

    env.execute()

  }
}
